package com.atguigu.stream.transale.withstatewindow

import com.atguigu.stream.util.MyApp
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * description ：ReduceByKey带窗口
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/15 20:14
 * modified By ：
 * version:    : 1.0
 */
object ReduceByKeyAndWindow extends MyApp {


  override def readAndTranform(ssc: StreamingContext): Unit = {

    // 获取数据
    val source: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop101", 9999)

    // TODO　1、ReduceByKey带窗口 每次计算Seconds(9)个步长的数据

    //    source.transform {
    //      rdd =>
    //        rdd.flatMap(_.split("\\s+")).map((_, 1))
    //    }
    //      .reduceByKeyAndWindow(_+_, Seconds(9))
    //      .print()

    // TODO　1、ReduceByKey带窗口 优化，每次计算Seconds(9)个步长的数据其中有部分数据是以前的，所以不需要重新参与运算
    // TODO 这种优化手段，需要checkpoint
    ssc.checkpoint("./kafkaSource01")
    source.transform {
      rdd =>
        rdd.flatMap(_.split("\\s+")).map((_, 1))
    }
      // TODO _+_：上一次结果 + 本次步长个单位的结果 结果1
      // TODO _-_：结果1 - 最后一次步长个单位的结果
      // TODO 使用这种操作，最后会出现（a,0）的情况，所以需要添加过滤 filterFunc = kv => kv._2 != 0
      .reduceByKeyAndWindow(_ + _, _ - _, Seconds(9), filterFunc = kv => kv._2 != 0)
      .print()
  }

}
